<?php


namespace coco\queue;


use coco\App;
use coco\Config;
use coco\Log;

final class Master
{
    protected $config;
    private $worker = [];
    private $monitor_pid = 0;
    private $terminate = false; //是否中断
    private $queue;

    private function __construct(string $queue)
    {
        $this->queue = $queue;
        $this->config = self::config();
    }

    private function __clone()
    {
    }

    private static function config(): array
    {
        $config = Config::getInstance()->get('queue');
        return [
            'worker_num' => $config['worker_num'] ?? 1,
            'user' => $config['user'] ?? 'www',
            'pid_file_path' => App::getInstance()->runtime_path,
            'output' => $config['output'] ?? '/dev/null',
            'monitor_cd' => $config['monitor_cd'] ?? 60, //0表示不检测
        ];
    }

    private static function posix_kill($queue, $signo): bool
    {
        $pid_file_name = 'queue_' . $queue . '.pid';
        $pid_file = self::config()['pid_file_path'] . $pid_file_name;
        if (!is_file($pid_file)) {
            return false;
        }
        $pid = @file_get_contents($pid_file);
        if (empty($pid)) {
            return false;
        }
        if (!$running = posix_kill($pid, $signo)) {
            unlink($pid_file);
        }
        return $running;
    }

    public static function stop($queue)
    {
        if (!self::posix_kill($queue, SIGTERM)) {
            exit('没有运行' . PHP_EOL);
        }
    }

    public static function reload($queue)
    {
        if (!self::posix_kill($queue, SIGUSR1)) {
            exit('没有运行' . PHP_EOL);
        }
    }

    public static function restart($queue)
    {
        if (!self::posix_kill($queue, SIGTERM)) {
            exit('没有运行' . PHP_EOL);
        }
        do {
            coco_sleep(0.1);
        } while (self::posix_kill($queue, 0));

        self::start($queue);
    }

    public static function status($queue)
    {

        //todo
    }


    public static function start($queue)
    {
        (new self($queue))->run();
    }

    private function run()
    {
        $this->daemon();
        cli_set_process_title('coco queue ' . $this->queue . ' master start_file=' . dirname(App::getInstance()->app_path));
        $this->log('queue master start:' . posix_getpid());
        $this->initSignal();

        while (1) {
            try {
                coco_sleep(0.1);
                pcntl_signal_dispatch();
                if ($this->terminate) {
                    break;
                }
                $this->createMonitor();
                $this->createWorker();
                Job::getToRun($this->queue);
            } catch (\Throwable $e) {
            }
        }
        if (file_exists($this->config['pid_file_path'] . 'queue_' . $this->queue . '.pid')) {
            unlink($this->config['pid_file_path'] . 'queue_' . $this->queue . '.pid');
        }
        $this->log('will stop:' . posix_getpid());
        posix_kill($this->monitor_pid, SIGTERM);
        $worker_pid = array_values($this->worker);
        foreach ($worker_pid as $pid) {
            posix_kill($pid, SIGTERM);
        }
        while (count($this->worker) > 0) {
            pcntl_signal_dispatch();
            coco_sleep(1);
        }
        $this->log('stop success:' . posix_getpid());
        exit(0);
    }

    private function createWorker()
    {
        if (count($this->worker) < $this->config['worker_num']) {
            //这里释放掉redis 不然子进程报错
            Job::freeRedis();
            while (count($this->worker) < $this->config['worker_num']) {
                $pid = pcntl_fork();
                if ($pid > 0) {
                    $this->worker[$pid] = $pid;
                } elseif ($pid === 0) {
                    (new Worker($this->queue))->run();
                    exit(0);
                }
            }
        }
    }

    private function createMonitor()
    {
        if ($this->config['monitor_cd'] <= 0) {
            return;
        }
        //进程不存在
        if (empty($this->monitor_pid) || false === posix_kill($this->monitor_pid, 0)) {
            //这里释放掉redis 不然子进程报错
            Job::freeRedis();
            $pid = pcntl_fork();
            if ($pid > 0) {
                $this->monitor_pid = $pid;
            } elseif ($pid === 0) {
                (new Monitor($this->queue, ['cd' => $this->config['monitor_cd']]))->run();
                exit(0);
            }
        }
    }

    private function daemon()
    {
        if (self::posix_kill($this->queue, 0)) {
            echo "已经运行" . PHP_EOL;
            exit();
        }
        self::show_logo();
        umask(0); //把文件掩码清0
        // 一次fork
        if (pcntl_fork() != 0) { //是父进程，父进程退出
            exit();
        }
        posix_setsid(); //设置新会话组长，脱离终端
        if (pcntl_fork() != 0) { //是第一子进程，结束第一子进程
            exit();
        }
        chdir("/"); //改变工作目录
        $user = posix_getpwnam($this->config['user']);
        if ($user) {
            $uid = $user['uid'];
            $gid = $user['gid'];
            if (false === posix_setuid($uid)) {
                echo "设置运行用户出错" . PHP_EOL;
                exit;
            }
            posix_setgid($gid);
        } else {
            echo "不存在的运行用户" . PHP_EOL;
            exit;
        }
        //关闭打开的文件描述符
        fclose(STDIN);
        fclose(STDOUT);
        fclose(STDERR);
        global $stdin, $stdout, $stderr;
        $stdin = fopen($this->config['output'], 'r');
        $stdout = fopen($this->config['output'], 'a');
        $stderr = fopen($this->config['output'], 'a');
        file_put_contents($this->config['pid_file_path'] . 'queue_' . $this->queue . '.pid', posix_getpid());
    }


    private function initSignal()
    {
        pcntl_signal(SIGUSR1, [$this, 'signalHandler']);
        pcntl_signal(SIGCHLD, [$this, 'signalHandler']);
        pcntl_signal(SIGTERM, [$this, 'signalHandler']);
        pcntl_signal(SIGHUP, [$this, 'signalHandler']);
        pcntl_signal(SIGQUIT, [$this, 'signalHandler']);
    }

    //信号处理函数
    private function signalHandler($signo)
    {
        switch ($signo) {
                //重启
            case SIGUSR1:
                $worker_pid = array_values($this->worker);
                foreach ($worker_pid as $pid) {
                    posix_kill($pid, SIGTERM);
                }
                break;
                //子进程结束信号
            case SIGCHLD:
                while (($pid = pcntl_waitpid(-1, $status, WNOHANG)) > 0) {
                    unset($this->worker[$pid]);
                }
                break;
                //中断进程
            case SIGTERM:
            case SIGHUP:
            case SIGQUIT:
                $this->terminate = true;
                break;
            default:
        }
    }

    private function log($msg)
    {
        Log::getInstanceByKey('queue', [['sub_path' => 'queue/']])->write('queue ' . $this->queue . ' master :' . $msg);
    }

    private static $is_showed_logo = false;

    private static function show_logo()
    {
        if (!self::$is_showed_logo) {
            self::$is_showed_logo = true;
            echo '
▄████████  ▄██████▄   ▄████████  ▄██████▄  
███    ███ ███    ███ ███    ███ ███    ███ 
███    █▀  ███    ███ ███    █▀  ███    ███ 
███        ███    ███ ███        ███    ███ 
███        ███    ███ ███        ███    ███ 
███    █▄  ███    ███ ███    █▄  ███    ███ 
███    ███ ███    ███ ███    ███ ███    ███ 
████████▀   ▀██████▀  ████████▀   ▀██████▀  
                                            
';
        }
    }
}
